package tk.xboot.flink.kfk.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class MessageSplitter implements FlatMapFunction<String, Tuple2<String,Long>> {
    public static final String SPLITTER = ",";
    /**
     * The core method of the FlatMapFunction. Takes an element from the input data set and transforms
     * it into zero, one, or more elements.
     *
     * @param value The input value.
     * @param out   The collector for returning result values.
     * @throws Exception This method may throw exceptions. Throwing an exception will cause the operation
     *                   to fail and may trigger recovery.
     */
    @Override
    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
        if (value != null && value.contains(SPLITTER)) {
            String[] parts = value.split(SPLITTER);
            out.collect(new Tuple2<>(parts[1],Long.parseLong(parts[2])));
        }
    }
}
